package com.zhisheng.sql.blink.stream.example;

import com.google.gson.Gson;
import com.zhisheng.sql.blink.stream.example.websocket.PathBasedSocketServer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.ArrayList;
import java.util.List;

public class WebSocketSink extends RichSinkFunction<CasinoWinJob_test.SumWin> {
    private static volatile PathBasedSocketServer pathBasedSocketServer;

    @Override
    public void open(Configuration parameters) throws Exception {
        if (pathBasedSocketServer == null){
            synchronized (WebSocketSink.class){
                if (pathBasedSocketServer == null){
                    pathBasedSocketServer = new PathBasedSocketServer(9000);
                    pathBasedSocketServer.start();
                }
            }
        }

    }

    @Override
    public void invoke(CasinoWinJob_test.SumWin value, Context context) throws Exception {
        //解析出path
        String path = value.getPath();
        pathBasedSocketServer.sendMessage(new Gson().toJson(value),path);

        List<String> bads = new ArrayList<>();
        bads.add("1");
        bads.add("2");
        bads.add("3");
        bads.add("4");
    }


    @Override
    public void close() throws Exception {

    }




}